package south

import (
	"errors"
	"iot-base/common/types"
	"reflect"
	//"fmt"
	//"iot-baes/common/token"
	"encoding/json"
	"iot-base/common/cache"
	"iot-base/common/config"
	"iot-base/common/ctrltype"
	"iot-base/common/dbobj"
	"iot-base/common/event"
	"iot-base/common/eventtype"
	"iot-base/common/maintain"
	"iot-base/common/mqtt"
	"iot-base/common/notify"
	//"iot-base/common/notify"

	//"iot-base/common/webapi"
	"fmt"
	"math"
	"time"
)

const DBError = -330001

var client mqtt.IOTMQTTBroker

//var upGradeChan chan []string

func init() {
	//refreshOption()
	if !config.MQTTOption.IsMqttConnection {
		return
	}
	notify.SubMsg(true, "upgrade_edges", &maintain.UpgradeCmd{}, func(t string, body interface{}) {
		if body != nil {
			pCmd, _ := body.(*maintain.UpgradeCmd)
			if pCmd != nil {
				if pCmd.URL != "" {
					if len(pCmd.EdgeIDs) == 1 && pCmd.EdgeIDs[0] == "all" {
						upgradeEdge("all", pCmd.URL, pCmd.Param)
					} else {
						for _, id := range pCmd.EdgeIDs {
							upgradeEdge(id, pCmd.URL, pCmd.Param)
						}
					}
				}
			}
		}
	})
}

func upgradeEdge(id string, url string, param string) {
	cmd := &mqtt.Cmd{
		CmdType: "upgrade",
		Body: maintain.UpgradeCmd{
			URL:   url,
			Param: param,
		},
	}
	client.SendCmd(cmd, config.GetEdgeTopic()+"upgrade/"+id, 0, nil, nil, nil)
}

func handleShakeHand(client mqtt.IOTMQTTBroker, topic string, pCmd *mqtt.Cmd, isRigester bool) {
	if pCmd != nil {

		//logger.Log.Info("get hb/register-------------------------")
		now := time.Now().Unix()
		if math.Abs(float64(now-pCmd.TimeStamp)) > 300 {
			//fmt.Println("timeout hb/regiester-----------", float64(now-pCmd.TimeStamp))
			return
		}

		isExist, err := cache.IsInThingID(pCmd.SourceID)
		if isExist && err == nil {
			cache.RefreshSvrOption()
			client.Reply(config.GetEdgeTopic(), pCmd, 0, "", cache.GetSvrOption())
			fmt.Println("get--------register---------", pCmd.SourceID, config.GetEdgeTopic(), cache.GetSvrOption().HttpOuterAddr)
			lastTime, _ := cache.GetLastAliveTime(pCmd.SourceID)
			if now-lastTime > 120 || isRigester {
				event.ClearOfflineAlarm(pCmd.SourceID)
				fmt.Println("-----------keep step alarm----------------")
				keepStepAlarm(pCmd.SourceID)
			}
			cache.SetLastAliveTime(pCmd.SourceID, now)

		} else {
			//fmt.Println("error regist--------------------", pCmd.SourceID, err)
			client.Reply(config.GetEdgeTopic(), pCmd, -1, "not exist "+pCmd.SourceID, nil)
		}
	}
}

func handleRegister(client mqtt.IOTMQTTBroker, topic string, pCmd *mqtt.Cmd) {
	handleShakeHand(client, topic, pCmd, true)
}

func handleHB(client mqtt.IOTMQTTBroker, topic string, pCmd *mqtt.Cmd) {
	handleShakeHand(client, topic, pCmd, false)
}

/*
func handleQuery(client mqtt.IOTMQTTBroker, topic string, pCmd *mqtt.Cmd) {
	if pCmd != nil && pCmd.Body != nil {
		query, ok := pCmd.Body.(edgeprotocol.Query)
		if ok {
			if query.Check() {
				sql := "select * from " + query.QueryType
				if query.Condition != "" {
					sql = sql + " where " + query.Condition
				}
				var result []map[string]interface{}
				err := dbconn.DBConnection.Raw(sql).Find(&result).Error
				if err != nil {
					client.Reply("query", pCmd, DBError, err.Error(), nil)
				} else {
					client.Reply("query", pCmd, 0, "", result)
				}
			}

		}
	}
}
*/
func handleDataReport(client mqtt.IOTMQTTBroker, topic string, pCmd *mqtt.Cmd) {
	if pCmd != nil {
		//fmt.Println("-------------------topic run----------handleDataReport-------", pCmd.Body)
		pData, ok := pCmd.Body.(map[string]interface{})
		if ok {

			tid, _ := pData["tid"].(string)
			pid, _ := pData["pid"].(string)

			sid, _ := pData["sub_id"].(string)
			mmi, oki := pData["mindex"]
			//fmt.Println("data comming-----222222222-------", mmi, reflect.TypeOf(mmi))
			var mi = 0
			if oki {
				mf, okf := mmi.(float64)
				if okf {
					mi = int(mf)
				} else {
					mf32, okf32 := mmi.(float32)
					if okf32 {
						mi = int(mf32)
					} else {
						mi, oki = mmi.(int)
						if !oki {
							i64mi, ok64 := mmi.(int64)
							if ok64 {
								mi = int(i64mi)
							}
						}
					}
				}
			}
			rt, _ := pData["rt"].(int64)
			pt, _ := pData["pt"].(int64)
			if pid == "streams" && sid == "token" {
				fmt.Println("data comming------------", pData)
				fmt.Println("data comming-----222222222-------", mi, pid, sid, pData["v"])
			}

			notify.RawData(tid, pid, sid, mi, pData["v"], rt, pt)
		}
	}

}

/*
func handelEmpty(client mqtt.IOTMQTTBroker, topic string, pCmd *mqtt.Cmd) {
	if pCmd != nil {
		//fmt.Println("-------------------topic run----------handleDataReport-------", pCmd.Body)
		fmt.Println("-----------------get msg-------------", topic, pCmd)
	}

}*/

//InitDataPicker 初始化picker
func InitDataPicker() {
	//mqtt.RegisterWithPattern("", topicDataReport, handleData)
	//mqtt.CreateClient("")
	if !config.MQTTOption.IsMqttConnection {
		return
	}
	client = mqtt.CreateMQTTClient(config.MQTTOption.Addr, config.MQTTOption.UN, config.MQTTOption.PWD, true,
		config.SelfOption.ID)

	client.RegisterTopic("$queue/"+config.GetCloudTopic()+"register/+", handleRegister, nil)
	client.RegisterTopic("$queue/"+config.GetCloudTopic()+"hb/+", handleHB, nil)
	client.RegisterTopic("$queue/"+config.GetCloudTopic()+"data/+", handleDataReport, nil)
	client.RegisterTopic("$queue/"+config.GetCloudTopic()+"alarm/+", handleAlarm, &eventtype.Alarm{})
	client.RegisterTopic("$queue/"+config.GetCloudTopic()+"event/+", handleEvent, &eventtype.Event{})
	client.RegisterTopic("$queue/"+config.GetCloudTopic()+"state/+", handleState, &[]dbobj.ThingState{})
	client.RegisterTopic("$queue/"+config.GetCloudTopic()+"record/+", handleRecord, &types.RecordData{})

	notify.SubCtrlCmd("IOTBASEmEdgeCTL",
		func(pCmd *ctrltype.CtrlCmd) (map[string]interface{}, error) {
			if pCmd != nil { //云端，发送给边缘侧
				cmd := &mqtt.Cmd{
					CmdType: "remoteCtrl",
					Body:    pCmd,
				}
				var id = pCmd.ThingID
				if pCmd.EdgeID != "" {
					id = pCmd.EdgeID
				}
				err, istimeout, body := client.SendCmdWait(cmd, config.GetEdgeTopic()+"remoteCtrl/"+id, 10)
				var mapV map[string]interface{}
				if body != nil {
					bbody, _ := body.(*json.RawMessage)
					if bbody != nil {
						err = json.Unmarshal(*bbody, &mapV)
					} else {
						mapV, _ = body.(map[string]interface{})
						fmt.Println("----------body type :", reflect.TypeOf(body))
					}

				}
				if err == nil && istimeout {
					err = errors.New("timeout error")
				}
				return mapV, err
			}
			return nil, nil
		})

}
